package com.credithc.signature.core.rabbitmq;


import com.credithc.signature.common.ro.core.CoreRequet;
import com.credithc.signature.core.service.CoreHandler;
import com.credithc.signature.core.util.ProtobufUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 描述：消费者监听，不用绑死队列名<br/>
 * 作者：sunyang
 * 创建：2019/6/3 14:27
 * 版本：v1.0.0<br/>
 */
@Slf4j
@Component
public class RabbitMQProcess implements ChannelAwareMessageListener {

    @Autowired
    private CoreHandler coreHandler;

    private static final String MDC_REQUEST_SN = "mdc_request_sn";

    ThreadPoolExecutor threadPoolExecutor;

    public RabbitMQProcess() {
        this.threadPoolExecutor = new ThreadPoolExecutor(5, 5, 30, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        CoreRequet coreRequet = null;
        try {
            coreRequet = ProtobufUtil.deserializer(message.getBody(), CoreRequet.class);
            log.info("－消息－: requestSn={}", coreRequet.getRequestSn(), coreRequet.toString());
            try {
                MDC.put(MDC_REQUEST_SN, coreRequet.getRequestSn());
            } catch (Exception e) {
                // 不做处理
                log.error("MDC添加发生异常......", e.getMessage());
            }

            // 处理逻辑
            log.info("－消息－: requestSn={},任务总数:{},已完成的任务数量:{},正在执行的任务数量:{},队列中任务数量:{},线程池当前的线程数量:{},线程池曾经最大线程数量:{}", coreRequet.getRequestSn(), threadPoolExecutor.getTaskCount(),
                    threadPoolExecutor.getCompletedTaskCount(), threadPoolExecutor.getActiveCount(), threadPoolExecutor.getQueue().size(), threadPoolExecutor.getPoolSize(), threadPoolExecutor.getLargestPoolSize());
            this.threadPoolExecutor.execute(new SignatureTask(coreRequet));
        } catch (Throwable e) {
            if (coreRequet != null) {
                log.error("－消息－: requestSn={},处理失败发回mq", coreRequet.getRequestSn());
            } else {
                log.error("消息处理失败发回mq", e);
            }
        } finally {
            try {
                // 确认消息成功接收
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Throwable e) {
                // 不做处理
                log.error("消息确认发生异常......", e.getMessage());
            }
            try {
                MDC.remove(MDC_REQUEST_SN);
            } catch (Exception e) {
                // 不做处理
                log.error("MDC删除发生异常......", e.getMessage());
            }
        }
    }


    class SignatureTask implements Runnable, Serializable {
        CoreRequet coreRequet;

        SignatureTask(CoreRequet coreRequet) {
            this.coreRequet = coreRequet;
        }

        @Override
        public void run() {
            coreHandler.handle(coreRequet);
        }
    }

}



